/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

package maxent.mr.trainer;

import maxent.mr.mapred.DictionaryMapper;
import maxent.mr.mapred.DictionaryReducer;
import maxent.mr.writable.TagCountWritable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

/**
 *
 * Confguration and runner for Dictionary MapReduce
 * 
 * @author hadoop
 */
public class DictionaryRunner extends Configured implements Tool{

    public int run(String[] args) throws Exception {

        if(args.length != 2){
            System.err.println("Usage: <input path> <tagseparator>");
            System.exit(-1);
        }


        String inputDir = args[0];
        String outputDir = inputDir+"-dictionary";
        String tagseparator = args[1];
        
        Configuration conf = new Configuration();
        conf.set("tagseparator", tagseparator);

        // setting bigger heap space
        //conf.set( "mapred.child.java.opts", "-Xmx2000m");

        Job job = new Job(conf);
        job.setJarByClass(DictionaryRunner.class);

        FileInputFormat.addInputPath(job, new Path(inputDir));
        FileOutputFormat.setOutputPath(job, new Path(outputDir));

        job.setMapperClass(DictionaryMapper.class);
        job.setReducerClass(DictionaryReducer.class);

        job.setInputFormatClass(TextInputFormat.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        job.setOutputFormatClass(TextOutputFormat.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(TagCountWritable.class);

        return job.waitForCompletion(true) ? 0 : 1;
        
    }

    public static void main(String[]args) throws Exception{

        System.out.println("Starting Job Dictionary");
        long startTime = System.currentTimeMillis();
        int res = ToolRunner.run(new Configuration(), new DictionaryRunner(), args);
        System.out.println("Job Dictionary finished in " + (System.currentTimeMillis()-startTime)/1000.0 + " seconds");
        System.exit(res);

    }

}
